#!bin/bash

if [ -n "$1" ] ; then
  data_date=$1
else
  data_date=`date -d '-1 days' +%F`
fi

TMP_ADS_PARSE_INFO_SQL="
DROP TABLE IF EXISTS jtp_ads_warehouse.tmp_ads_parse_info;
CREATE TABLE IF NOT EXISTS jtp_ads_warehouse.tmp_ads_parse_info
AS
SELECT
    request_uri
    ,parse_url('https://www.baidu.com' || request_uri,'QUERY','t') AS event_time
    ,split(parse_url('https://www.baidu.com' || request_uri,'PATH'),'/')[2] AS platform_name_en
    ,split(parse_url('https://www.baidu.com' || request_uri,'PATH'),'/')[3] AS event_type
    ,parse_url('https://www.baidu.com' || request_uri,'QUERY','id') AS ad_id
    ,parse_url('https://www.baidu.com' || request_uri,'QUERY','ip') AS client_ip
    ,parse_url('https://www.baidu.com' || request_uri,'QUERY','device_id') AS client_device_id
    ,parse_url('https://www.baidu.com' || request_uri,'QUERY','os_type') AS client_os_type
    ,reflect('java.net.URLDecoder','decode',parse_url('https://www.baidu.com' || request_uri,'QUERY','ua'),'utf-8') AS client_ua
FROM jtp_ads_warehouse.ods_ads_log_inc
;
"
TMP_ADS_LOG_DIM_SQL="
DROP TABLE IF EXISTS jtp_ads_warehouse.tmp_ads_log_dim;
CREATE TABLE IF NOT EXISTS jtp_ads_warehouse.tmp_ads_log_dim
AS
SELECT
    t1.event_time
    ,t1.event_type
    ,t1.ad_id AS ads_id
    ,t2.ad_name AS ads_name
    ,t2.product_id AS ads_product_id
    ,t2.product_name AS ads_product_name
    ,t2.product_price AS ads_product_price
    ,t2.material_id AS ads_material_id
    ,t2.ad_group_id
    ,t2.platform_id
    ,t2.platform_name AS platform_name_en
    ,t2.platform_alias_zh AS platform_name_zh
    ,t1.client_ip
    ,t1.client_device_id
    ,t1.client_os_type
    ,t1.client_ua AS client_user_agent
FROM jtp_ads_warehouse.tmp_ads_parse_info t1
    LEFT JOIN (
        SELECT
            id, ad_id, ad_name, ad_group_id, product_id, product_name, product_price,
            material_id, platform_id, platform_name, platform_alias_zh, create_time, cancel_time
        FROM jtp_ads_warehouse.dim_ads_platform_indo_full
        WHERE dt = '${data_date}'
    )t2 ON t1.ad_id = t2.ad_id
;
"
TMP_ADS_LOG_REGION_SQL="
DROP TABLE IF EXISTS jtp_ads_warehouse.tmp_ads_log_region;
CREATE TABLE IF NOT EXISTS jtp_ads_warehouse.tmp_ads_log_region
AS
SELECT
    event_time
    ,event_type
    ,ads_id
    ,ads_name
    ,ads_product_id
    ,ads_product_name
    ,ads_product_price
    ,ads_material_id
    ,ad_group_id AS ads_group_id
    ,platform_id
    ,platform_name_en
    ,platform_name_zh
    ,client_map['country'] AS client_country
    ,client_map['area'] AS client_area
    ,client_map['province'] AS client_province
    ,client_map['city'] AS client_city
    ,client_ip
    ,client_device_id
    ,client_os_type
    ,client_user_agent
FROM (
         SELECT
             *,jtp_ads_warehouse.ip_to_region(client_ip) AS client_map
         FROM jtp_ads_warehouse.tmp_ads_log_dim
);
"

TMP_ADS_LOG_BROWSER_SQL="
DROP TABLE IF EXISTS jtp_ads_warehouse.tmp_ads_log_browser;
CREATE TABLE IF NOT EXISTS jtp_ads_warehouse.tmp_ads_log_browser
AS
SELECT event_time,
       event_type,
       ads_id,
       ads_name,
       ads_product_id,
       ads_product_name,
       ads_product_price,
       ads_material_id,
       ads_group_id,
       platform_id,
       platform_name_en,
       platform_name_zh,
       client_country,
       client_area,
       client_province,
       client_city,
       client_ip,
       client_device_id,
       client_os_type,
       client_ua_map['os_version'] AS client_os_version,
       client_ua_map['browser'] AS client_browser_type,
       client_ua_map['browser_version'] AS client_browser_version,
       client_user_agent
FROM (SELECT *,
             jtp_ads_warehouse.ua_to_browser(client_user_agent) AS client_ua_map
      FROM jtp_ads_warehouse.tmp_ads_log_region) t1
;
"

TMP_ADS_LOG_TRAFFIC_SQL="
DROP TABLE IF EXISTS jtp_ads_warehouse.tmp_ads_log_traffic;
CREATE TABLE IF NOT EXISTS jtp_ads_warehouse.tmp_ads_log_traffic
AS
WITH
    tmp_client AS (
        SELECT
            DISTINCT client_ip
        FROM (
                 SELECT
                     client_ip,ads_id,event_time
                      ,count(1) OVER(PARTITION BY client_ip,ads_id ORDER BY CAST(event_time AS BIGINT)
                     RANGE BETWEEN 300000 PRECEDING AND CURRENT ROW
                     ) AS cnt
                 FROM jtp_ads_warehouse.tmp_ads_log_browser
             )t1
        WHERE cnt > 100

        UNION

        SELECT
            DISTINCT client_ip
        FROM (
                 SELECT
                     client_ip, ads_id, event_time, event_point
                      ,(event_point - event_time) AS diff_time
                 FROM (
                          SELECT
                              client_ip,ads_id,event_time
                               ,lead(event_time,1,0) OVER(PARTITION BY client_ip,ads_id ORDER BY event_time) AS event_point
                          FROM jtp_ads_warehouse.tmp_ads_log_browser
                      )t1
             )t2
        GROUP BY client_ip,ads_id,diff_time
        HAVING count(1) > 5
    ),
    tmp_client_device AS (
        SELECT
            DISTINCT client_device_id
        FROM (
                 SELECT
                     client_device_id,ads_id,event_time
                      ,COUNT(1) OVER(PARTITION BY client_device_id,ads_id ORDER BY CAST(event_time AS BIGINT)
                     RANGE BETWEEN 300000 PRECEDING AND CURRENT ROW
                     )  AS cnt
                 FROM jtp_ads_warehouse.tmp_ads_log_browser
             )t1
        WHERE cnt > 100

        UNION

        SELECT
            DISTINCT client_device_id
        FROM (
                 SELECT
                     client_device_id, ads_id, event_time, event_point
                      ,(event_point - event_time) AS diff_time
                 FROM (
                          SELECT
                              client_device_id,ads_id,event_time
                               ,lead(event_time,1,0) over (PARTITION BY client_device_id,ads_id ORDER BY event_time) AS event_point
                          FROM jtp_ads_warehouse.tmp_ads_log_browser
                      )t1
             )t2
        GROUP BY client_device_id,ads_id,diff_time
        HAVING count(1) > 5
    )
SELECT event_time,
       event_type,
       ads_id,
       ads_name,
       ads_product_id,
       ads_product_name,
       ads_product_price,
       ads_material_id,
       ads_group_id,
       platform_id,
       platform_name_en,
       platform_name_zh,
       client_country,
       client_area,
       client_province,
       client_city,
       t2.client_ip,
       t3.client_device_id,
       client_os_type,
       client_os_version,
       client_browser_type,
       client_browser_version,
       client_user_agent,
       t2.client_ip IS NULL AND t3.client_device_id IS NULL AS is_invalid_traffic
FROM jtp_ads_warehouse.tmp_ads_log_browser t1
    LEFT JOIN tmp_client t2 ON t1.client_ip = t2.client_ip
    LEFT JOIN tmp_client_device t3 ON t1.client_device_id = t3.client_device_id
;
"

DWD_ADS_EVENT_LOG_INC_SQL="
INSERT OVERWRITE TABLE jtp_ads_warehouse.dwd_ads_event_log_inc PARTITION (dt = '${data_date}')
SELECT CAST(event_time AS BIGINT),
       event_type,
       CAST(ads_id AS BIGINT),
       ads_name,
       ads_product_id,
       ads_product_name,
       ads_product_price,
       ads_material_id,
       ads_group_id,
       platform_id,
       platform_name_en,
       platform_name_zh,
       client_country,
       client_area,
       client_province,
       client_city,
       client_ip,
       client_device_id,
       client_os_type,
       client_os_version,
       client_browser_type,
       client_browser_version,
       client_user_agent,
       is_invalid_traffic
FROM jtp_ads_warehouse.tmp_ads_log_traffic
;
"

/opt/module/spark/bin/beeline -u jdbc:hive2://node101:10001 -n bwie -e "${TMP_ADS_PARSE_INFO_SQL}${TMP_ADS_LOG_DIM_SQL}${TMP_ADS_LOG_REGION_SQL}${TMP_ADS_LOG_BROWSER_SQL}${TMP_ADS_LOG_TRAFFIC_SQL}${DWD_ADS_EVENT_LOG_INC_SQL}"
